#include "MQTTClient.h"
#include <struct.h>
#include <time.h>

#define ADDRESS              "tcp://192.168.31.161:1883"
#define CLIENTID             "ExampleClientSub"
#define TOPIC_DATA           "/app/data/down"//接收  数据指令
#define TOPIC_CONTROL        "/app/control/down"//接收  控制指令
#define TOPIC_MODE           "/app/mode/down"//接收  模式指令
#define TOPIC_P_DATA         "/app/data/up"//回复  数据json
#define TOPIC_P_CONTROL      "/app/control/up"//回复  控制结果
#define TOPIC_P_MODE         "/app/mode/up"//  模式修改结果
#define QOS         1

void mqtt_init();//MQTT初始化
int cjson_table(int*t);//解析点表
void variety_report(struct std_node *p,struct std_node *p1); //变化上报函数
void initiative_report(struct std_node *p,struct std_node *p1); //主动上报函数
int control_agreement(char *buf);
int modify_table(int type,int period);
int modify_mode(char *buf);


time_t timi;
int t;//上报周期
struct tm* tm_t;
struct need  context;
static struct shm_param para;//共享内存相关
volatile MQTTClient_deliveryToken deliveredtoken;
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_message pubmsg = MQTTClient_message_initializer;      
MQTTClient_deliveryToken token;   

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)//备注：message->payload代表JSON字符串
{//接收消息
    struct need* overall=(struct need* )context;
	
    if(strcmp(topicName,"/app/data/down")==0)
    {
        printf("数据刷新语句块\n");
        initiative_report(overall->p_equipment,overall->p_edge);
    }
    else if(strcmp(topicName,"/app/control/down")==0)
    {
        printf("控制模块语句块\n");
	    control_agreement(message->payload);
    }
    else if(strcmp(topicName,"/app/mode/down")==0)
    {
        printf("修改模式语句块\n");
		modify_mode(message->payload);
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

int main()
{
    int table_ret=cjson_table(&t);//解析点表 
    //设备采集进程共享内存
    int shm_ret = -1;
    shm_ret = shm_init(&para, "shm_equipment", 1024);
    if(shm_ret < 0)
    {
        return -1;
    }
    context.p_equipment = shm_getaddr(&para);//内存首地址指针
    if(context.p_equipment == NULL)
    {
        return -1;
    }
    puts("设备采集进程内存创建完成");
    //边缘采集进程共享内存
    shm_ret = shm_init(&para, "shm_edge", 1024);
    if(shm_ret < 0)
    {
        return -1;
    }
    context.p_edge = shm_getaddr(&para);//内存首地址指针
    if(context.p_edge== NULL)
    {
        return -1;
    }
    puts("边缘采集进程内存创建完成");
    mqtt_init(&context);//MQTT初始化


    if(table_ret==1)
    {
        while(1)
        {
            printf("变化上报语句块\n");
			variety_report(context.p_equipment,context.p_edge);
            sleep(1);
        }
    }
    else if(table_ret==2)
    {
        while(1)
        {
            printf("周期上报语句块,上报周期为%d s\n",t);
            initiative_report(context.p_equipment,context.p_edge);

            sleep(t);
        }
    }
    else if(table_ret==0)
    {
        while(1)
        {

        }
    }
    shm_del(&para);//解除两块共享内存
    return 0;
}

void mqtt_init(struct need * context)
{
    int rc;
    MQTTClient_create(&client, ADDRESS, CLIENTID,//客户端创建
    MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client,context,NULL, msgarrvd, NULL);//描述符，断联，消息到达，分离

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    MQTTClient_subscribe(client, TOPIC_DATA, QOS);//订阅主题
    MQTTClient_subscribe(client, TOPIC_CONTROL, QOS);//订阅主题
    MQTTClient_subscribe(client, TOPIC_MODE, QOS);//订阅主题
	printf("mqtt初始化完成\n");
}

int cjson_table(int *t)
{
    //读点表文件
    int fd = open("./table.txt",O_RDONLY,0666);
    if(fd == -1){
        perror("open failure");
        return -1;
    }
    char buf[2048];
    memset(buf,0,sizeof(buf));
    ssize_t read_ret = read(fd,buf,sizeof(buf));
    if(read_ret==-1)
    {
        perror("read:");
        return -1;
    }
    //printf("%s",buf);//test
    //解析点表内容判断是变化上报还是周期上报
    cJSON *root = NULL;
    cJSON *item = NULL;
    cJSON *arry = NULL;
    root = cJSON_Parse(buf);
    if(!root)
    {
        printf("Error before:[%s]\n",cJSON_GetErrorPtr());
    }
    //获取设备上报属性
    item = cJSON_GetObjectItem(root,"report");
    arry = cJSON_GetObjectItem(item,"type");//解析type的值
    //printf("%s\n",cJSON_Print(arry));
    *t=atoi(cJSON_Print(cJSON_GetObjectItem(item,"period")));//解析上报周期
    close(fd);//关闭文件,防止i影响下次文件使用
    return atoi(cJSON_Print(arry));
}


void initiative_report(struct std_node *p,struct std_node *p1)//主动上报函数  读内存，构造JSON，发送JSON  ,p jiji  6  ,p1 niuniu 5
{   
    char sql[128]={0};
    char arr[12]={0};
    time(&timi);
    tm_t=localtime(&timi);
    cJSON * arry[6];
    cJSON * arry1[5];
    cJSON *root_r = cJSON_CreateObject();
    cJSON *item_i = cJSON_CreateObject();
    cJSON_AddItemToObject(root_r,"type",cJSON_CreateNumber(1));//1 采集回复指令
    cJSON_AddItemToObject(root_r,"result",cJSON_CreateNumber(0));//成功返回0 失败返回1
    cJSON *arry_a = cJSON_CreateArray();//创建数组
    cJSON_AddItemToObject(root_r,"data",arry_a);//加入
    for(int i=0;i<6;i++)
    {
        arry[i] = cJSON_CreateObject();
        cJSON_AddItemToArray(arry_a,arry[i]);
        cJSON_AddNumberToObject(arry[i],"key",p[i].key);
        if(p[i].key==1||p[i].key==2)
        {
            memset(arr,0,sizeof(arr));
            sprintf(arr,"%0.2f",p[i].new_val.f_val);
            cJSON_AddStringToObject(arry[i],"val",arr);
        }
        else
        {
            memset(arr,0,sizeof(arr));
            sprintf(arr,"%d",p[i].new_val.i_val);
            cJSON_AddStringToObject(arry[i],"val",arr);
        }
    }  
    for(int i=0;i<5;i++)
    {
        arry1[i] = cJSON_CreateObject();
        cJSON_AddItemToArray(arry_a,arry1[i]);
        cJSON_AddNumberToObject(arry1[i],"key",p1[i].key);
        if(p1[i].key==101||p1[i].key==102||p1[i].key==104)
        {
            memset(arr,0,sizeof(arr));
            sprintf(arr,"%0.2f",p1[i].new_val.f_val);
            cJSON_AddStringToObject(arry1[i],"val",arr);
        }
        else
        {
            memset(arr,0,sizeof(arr));
            sprintf(arr,"%d",p1[i].new_val.i_val);
            cJSON_AddStringToObject(arry1[i],"val",arr);
        }
    }
    //printf("%d 日 %d -%d -%d",tm_t->tm_mday,tm_t->tm_hour,tm_t->tm_min,tm_t->tm_sec);
    //发送JSON
    char buf[1024]={0}; 
    strcpy(buf,cJSON_Print(root_r)); 
    pubmsg.payload = buf;
	pubmsg.payloadlen = (int)strlen(buf);
	pubmsg.qos = QOS;
	pubmsg.retained = 0;
	MQTTClient_publishMessage(client, TOPIC_P_DATA , &pubmsg, &token);
}

void variety_report(struct std_node *p,struct std_node *p1)//变化上报函数
{
    char arr[12]={0};
    char sql[128]={0};
    time(&timi);
    tm_t=localtime(&timi);
    cJSON * arry[6];
    cJSON * arry1[5];     
    cJSON *root_r = cJSON_CreateObject();
    cJSON *item_i = cJSON_CreateObject();
    cJSON_AddItemToObject(root_r,"type",cJSON_CreateNumber(1));//1 采集回复指令
    cJSON_AddItemToObject(root_r,"result",cJSON_CreateNumber(0));//成功返回0 失败返回1
    cJSON *arry_a = cJSON_CreateArray();//创建数组
    cJSON_AddItemToObject(root_r,"data",arry_a);//加入
    for(int i=0;i<6;i++)
    {
        if(i<=1)
        {
            memset(arr,0,sizeof(arr));
            sprintf(arr,"%0.2f",p[i].new_val.f_val);
        }else
        {
            memset(arr,0,sizeof(arr));
            sprintf(arr,"%d",p[i].new_val.i_val); 
        }
        if(p[i].old_val.i_val==p[i].new_val.i_val)
        {
            continue;
        }
        else
        {
            arry[i] = cJSON_CreateObject();
            cJSON_AddItemToArray(arry_a,arry[i]);
            cJSON_AddNumberToObject(arry[i],"key",p[i].key);
            cJSON_AddStringToObject(arry[i],"val",arr);
        }
    }  
    for(int i=0;i<5;i++)
    {
        if(i<=1||i==3)
        {
            memset(arr,0,sizeof(arr));
            sprintf(arr,"%0.2f",p1[i].new_val.f_val);
        }else
        {
            memset(arr,0,sizeof(arr));
            sprintf(arr,"%d",p1[i].new_val.i_val); 
        }
        if(p1[i].old_val.i_val==p1[i].new_val.i_val)
        {
            continue;
        }
        else
        {
            arry1[i] = cJSON_CreateObject();
            cJSON_AddItemToArray(arry_a,arry1[i]);
            cJSON_AddNumberToObject(arry1[i],"key",p1[i].key); 
            cJSON_AddStringToObject(arry1[i],"val",arr);
        }  
    }
    //发送JSON
    char buf[1024]={0}; 
    strcpy(buf,cJSON_Print(root_r)); 
    pubmsg.payload = buf;
	pubmsg.payloadlen = (int)strlen(buf);
	pubmsg.qos = QOS;
	pubmsg.retained = 0;
	MQTTClient_publishMessage(client, TOPIC_P_DATA , &pubmsg, &token);
}

int control_agreement(char *buf)
{	
    char buff[128]={0};
	int key;
	struct msg_struct send,recv;
    cJSON *root = NULL;
    cJSON *item_type = NULL;
	cJSON *item_limit = NULL;
	cJSON *item_keys = NULL;
	cJSON *item_data = NULL;
	cJSON *item_data_key = NULL;
	cJSON *item_data_val = NULL;
    root = cJSON_Parse(buf);
    if(!root)
    {
        printf("Error before\n");
        return 0;
    }else
    {
        item_type = cJSON_GetObjectItem(root,"type");
		if(atoi(cJSON_Print(item_type)) == 1)
		{
			//采集指令
			item_limit = cJSON_GetObjectItem(root,"limit");
			printf("%s\n",cJSON_Print(item_limit));
			item_keys = cJSON_GetObjectItem(root,"keys");
			printf("%s\n",cJSON_Print(item_keys));
			
		}else
		{
			//控制指令
			item_data = cJSON_GetObjectItem(root,"data");
			printf("%s\n",cJSON_Print(item_data));
			item_data_key = cJSON_GetObjectItem(item_data,"key");
			key = atoi(cJSON_Print(item_data_key));
			item_data_val = cJSON_GetObjectItem(item_data,"val");

			strcpy(send.msg_val,cJSON_Print(item_data_val));
			send.key = key;
			if(key < 10)
			{
			//	printf("/////:%s\n",item_data_val);
				printf("控制指令发送\n");
				//发送给m0
				if(msg_queue_send("m0_msg", &send, sizeof(send), 0) < 0)
				{
					printf("msg_queue_send error\n");
					return -1;
				}
				if(msg_queue_recv("msg_report",&recv,sizeof(recv),0,0) < 0)
				{
					printf("msg_queue_recv_review error\n");

				}
				printf("接受到的回复消息%s\n",recv.msg_val);
			}else
			{
				//发送给mdbus
				if(msg_queue_send("mb_msg", &send, sizeof(send), 0) < 0)
				{
					printf("msg_queue_send error\n");
					return -1;
				}
				if(msg_queue_recv("msg_report",&recv,sizeof(recv),0,0) < 0)
				{
					printf("msg_queue_recv_review error\n");

				}
				printf("接受到的回复消息%s\n",recv.msg_val); 
			}
            strcpy(buff,recv.msg_val); 
            printf("接受到的回-------%s\n",buff); 
            pubmsg.payload = buff;
            pubmsg.payloadlen = (int)strlen(buff);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            printf("test\n");
            MQTTClient_publishMessage(client, TOPIC_P_CONTROL, &pubmsg, &token);
		}
    }

    return 0;
}
int modify_mode(char *buf)
{
	//读取json要修改成的模式
	int type,period;
    cJSON *QT_root = NULL;
	cJSON *QT_root_data = NULL;
    cJSON *QT_item_type = NULL;
	cJSON *QT_item_period = NULL;
    QT_root = cJSON_Parse(buf);
    if(!QT_root)
    {
        printf("Error before\n");
        return -1;
    }else
    {
		QT_root_data = cJSON_GetObjectItem(QT_root,"data");
        QT_item_type = cJSON_GetObjectItem(QT_root_data,"type");
		type = atoi(cJSON_Print(QT_item_type));
		QT_item_period = cJSON_GetObjectItem(QT_root_data,"period");
		period = atoi(cJSON_Print(QT_item_period));
		printf("进入文件修改模式\n");
		modify_table(type,period);
        //发送修改结果
        char buff[64]="{\n\"result\":0\n\"msg\":\"模式修改成功\"\n}";  
        pubmsg.payload = buff;
        pubmsg.payloadlen = (int)strlen(buff);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        printf("test\n");
        MQTTClient_publishMessage(client, TOPIC_P_MODE , &pubmsg, &token);
	}
	
	
    return 0;
}
int modify_table(int type,int period)
{
	int fd_r,fd_w;
	//读点表文件
    fd_r = open("./table.txt",O_RDONLY,0666);
    if(fd_r == -1)
	{
        perror("open failure");
        return -1;
    }
    char buf[2048];
    memset(buf,0,sizeof(buf));
    ssize_t read_ret = read(fd_r,buf,sizeof(buf));
    if(read_ret==-1)
    {
        perror("read:");
        return -1;
    }
    cJSON *root = NULL;
    cJSON *item = NULL;
	cJSON *item_type = NULL;
    root = cJSON_Parse(buf);
    if(!root)
    {
        printf("Error before:[%s]\n",cJSON_GetErrorPtr());
    }
	printf("读取点表文件完成\n");
    //获取设备上报属性
    item = cJSON_GetObjectItem(root,"report");
	//设置设备上报模式
	cJSON_ReplaceItemInObject(item,"type",cJSON_CreateNumber(type));
	cJSON_ReplaceItemInObject(item,"period",cJSON_CreateNumber(period));
	fd_w = open("./table.txt",O_WRONLY |O_TRUNC ,0666);
    if(fd_w == -1)
	{
        perror("open failure");
        return -1;
    } 
	memset(buf,0,sizeof(buf));
	stpcpy(buf,cJSON_Print(root));
    ssize_t write_ret = write(fd_w,buf,sizeof(buf));
    if(write_ret==-1)
    {
        perror("read:");
        return -1;
    }
	printf("写入点表文件完成\n");
	close(fd_r);
    close(fd_w);//关闭文件,防止i影响下次文件使用
    return 0;
}
